package operator.function.do_;

import io.reactivex.Notification;
import io.reactivex.Observable;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;
import util.Printer;
import util.RxJavaHelper;

class DoOnEachOperator {

    public static void main(String[] args) {
//        doOnEach1();
        doOnEach2();
    }

    static void doOnEach1(){
        Observable.just(1)
                .doOnEach(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable disposable) {
                        Printer.print("doOnEach onSubscribe");
                    }

                    @Override
                    public void onNext(Integer integer) {
                        Printer.print("doOnEach onNext integer = " + integer);
                    }

                    @Override
                    public void onError(Throwable throwable) {
                        Printer.print("doOnEach onError throwable = " + throwable);
                    }

                    @Override
                    public void onComplete() {
                        Printer.print("doOnEach onComplete");
                    }
                })
                .subscribe(RxJavaHelper.getConsumer());
    }


    static void doOnEach2(){
        Observable.just(1)
                .doOnEach(new Consumer<Notification<Integer>>() {
                    @Override
                    public void accept(Notification<Integer> notification) throws Exception {
                        Printer.print("doOnEach notification = " + notification);
                        if(notification.isOnNext()){
                            Printer.print("doOnEach notification.value = " + notification.getValue());
                        }
                    }
                })
                .subscribe(RxJavaHelper.getConsumer());
    }


}
